﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using Sustainalytics.Entities.ConnectionStrings;

using System.Collections.Concurrent;

using MongoDB.Driver;

using Sustainalytics.Utils;
using Sustainalytics.Utils.Configuration;

using System.IO;
using System.Threading;
using System.Diagnostics;
using System.Configuration;
using Sustainalytics.DataLayer;


namespace Sustainalytics.PregenerateScreeningReports.Service
{
    public class Orchestrator
    {
        string _clientDWMasterCnx;
        string _clientPlatformCnx;
        string _myInstanceId;
        string _redisServerAddress;
        string _esgComputationServiceAddress;
        string _tempPath;
        int _loadIndicator;
        int _pause;
        ScreeningToolReportDistributionSession _session;


        public static Orchestrator NewInstanceFromConfig()
        {
            var cnx1 = ConfigurationManager.ConnectionStrings["ClientDWMaster"].ConnectionString;
            var cnx2 = ConfigurationManager.ConnectionStrings["ClientPlatform"].ConnectionString;
            var id = SettingsHelper.Read<string>("AWSInstanceId", "developer");
            var redisAddr = SettingsHelper.Read<string>("RedisServer", "localhost");
            var esgAddr = SettingsHelper.Read<string>("ESGComputeURL", "http://localhost:10280/esgcomputingserv/api/v1/esgcompute");
            var tempPath = SettingsHelper.Read<string>("TempPath", System.IO.Path.GetTempPath());
            var load = SettingsHelper.Read<int>("MaxSubProcess", System.Environment.ProcessorCount);
            var pause = SettingsHelper.Read<int>("SpawnPause", 3000);

            return new Orchestrator(cnx1, cnx2, id, redisAddr, esgAddr, tempPath, load, pause);
        }

        public Orchestrator()
        {
        }

        public Orchestrator(string clientDWMasterCnx, string clientPlatformCnx, string instanceId, string redisServerAddress, 
                            string esgComputationServiceAddress, string tempPath, int loadIndicator, int pause)
        {
            PSREventSource.Log.Orchestrator( Process.GetCurrentProcess().Id,clientDWMasterCnx, clientPlatformCnx, instanceId, redisServerAddress,
                            esgComputationServiceAddress, tempPath, loadIndicator);

            _myInstanceId = instanceId;
            _clientDWMasterCnx = clientDWMasterCnx;
            _clientPlatformCnx = clientPlatformCnx;           
            _redisServerAddress = redisServerAddress;
            _esgComputationServiceAddress = esgComputationServiceAddress;
            _tempPath = tempPath;
            _loadIndicator = loadIndicator;
            _pause = pause;

            // search for lastest session            
            var sessionColl = MongoDatabaseFactory.GetDatabase(_clientDWMasterCnx)
                                                  .GetCollection<ScreeningToolReportDistributionSession>();

            var date = DateTime.UtcNow.Date;
            _session = sessionColl.AsQueryable()
                                .Where(s => s.Date >= date && s.Completed == false)
                                .OrderByDescending(s => s.Date)
                                .FirstOrDefault();
            if (_session == null)
                throw new PSR_NoSessionFound();         
            
            var myMachine = _session.Machines.FirstOrDefault(x => x.MachineId == _myInstanceId);                        
            if (myMachine == null)
                throw new PSR_NoMachineFoundInSession(instanceId, _session.Id);

            TTLManager.Update(TimeSpan.FromMinutes(myMachine.TTLMinutes));
        }

        public bool Run()
        {

            var cron = new Cronometer();

            //only on master machine
            if (_myInstanceId == _session.MainMachine)
            { // add logging
                var log = ScreeningReportsDistributionLogic.RunDistribution(_session, _clientDWMasterCnx, _clientPlatformCnx,
                                _session.ConnectionStrings[ConnectionStringType.ClientUniverses],
                                _session.ConnectionStrings[ConnectionStringType.ClientDW]);

                PSREventSource.Log.Distribution(log);
                cron.Mark("RunDistribution");
            }

            var cron2 = new Cronometer();
            var reports = GenerateMaximalReports(_session, _redisServerAddress, _esgComputationServiceAddress, _tempPath, cron2);
            cron.Mark("GenerateMaximalReports: " + string.Join("|", reports));
            cron.AddMarks(cron2);

            var jobs = GenerateJobsForRequests(reports, _session.Id, _myInstanceId, _clientDWMasterCnx, _loadIndicator);
            cron.Mark("GenerateJobsForRequests: " + string.Join(Environment.NewLine, jobs));

            File.WriteAllLines(Path.Combine(_tempPath, "PSRjobs.txt"), jobs);

            var text = SpawnSubprocess(jobs, _loadIndicator, Process.GetCurrentProcess().MainModule.FileName, _pause);
            cron.Mark("SpawnSubprocess: " + text);

            // write sessions as complete 
            UpdateSession(_session);
            PSREventSource.Log.OrchestratorRunComplete(Process.GetCurrentProcess().Id, cron.ToString());

            return true;
        }

        public List<string> GenerateMaximalReports(ScreeningToolReportDistributionSession session, string redisServerAddress, string esgComputationServiceAddress, string tempPath, Cronometer cron)
        {
            var result = new List<string>();
            var reportIds = _session.Machines.SelectMany(x => x.ReportIds).Distinct();
            var helper = new ScreeningToolHelper(session.ConnectionStrings, redisServerAddress, esgComputationServiceAddress);
            if (!Directory.Exists(tempPath))
                Directory.CreateDirectory(tempPath);

            cron.Mark("start");
            foreach(var r in reportIds)
            {
                try
                {                    
                    var filename = MaximalReportHelper.ComposeMaximalReportName(session.Id, r, tempPath);
                    helper.SaveMaximalReport(r, filename);
                    result.Add(filename);
                    cron.Mark(filename);
                }
                catch(Exception ex)
                {
                    PSREventSource.Log.Exception(ex);
                }
            }
            return result;
        }
        
        public IEnumerable<string> GenerateJobsForRequests(List<string> reportsNames,  Guid sessionId, string machineId, string clientDWMasterCnx, int loadIndicator)
        {   
            // load data - requests id and size
            var requestsColl = MongoDatabaseFactory.GetDatabase(clientDWMasterCnx)
                                                   .GetCollection<ScreeningToolReportRequest>();
            var requests = requestsColl.AsQueryable()
                                       .Where(x => x.SessionId == sessionId  && x.MachineId == machineId)
                                       .Select( x=> new Tuple<int, Guid>(x.CompaniesIds.Count(), x.Id))
                                       .ToList();

            // distribute  by requests size 
            var jobs  = new List<string>();
            var distribution  = ScreeningReportsDistributionLogic.EvenPartition<Guid>(loadIndicator, requests);            
            foreach(var repName in reportsNames)    // 4 reports 
            foreach (var sets in distribution)       // 8 cores                 
                jobs.AddRange(CommandLine(repName, sets));// compose command line arguments
            
            
            return jobs; 
        }

        public IEnumerable<string> CommandLine(string name, IEnumerable<Guid> guids, int limit = 1024)
        {
            var one         = "\"" + name + "\"" + " ";
            var nextLen     = ((new Guid()).ToString() + "|").Length;
            var remaining = (int)Math.Floor((double)(limit - one.Length) / (double)nextLen);

            var skip = 0;
            do
            {
                var ids = guids.Skip(skip).Take(remaining);
                if (ids != null && ids.Any())
                    yield return (one + string.Join("|", ids));
                else
                    break;
                
                skip += remaining;
            }
            while (skip < guids.Count());
        }


        public void UpdateSession(ScreeningToolReportDistributionSession session)
        {
            var sessionColl = MongoDatabaseFactory.GetDatabase(_clientDWMasterCnx)
                                                  .GetCollection<ScreeningToolReportDistributionSession>();

            session.Completed = true;
            sessionColl.ReplaceOne(Builders < ScreeningToolReportDistributionSession >.Filter.Eq(x=>x.Id ,session.Id),  session);
        }

        public string  SpawnSubprocess(IEnumerable<string> jobs, int loadindicator, string processName, int pause = 5000)
        {
            var log = new StringBuilder("ProcessName: " + processName);
            var dic = new ConcurrentDictionary<Process, string>();
          
            try
            {
                var stopper = new ManualResetEvent(true);
                var counter = 0;
                foreach (var cmdline in jobs)
                {
                    stopper.WaitOne();                    
                    Interlocked.Increment(ref counter);
                    if (counter >= loadindicator)
                        stopper.Reset();

                    var p = Process.Start(processName, cmdline);                    
                    p.EnableRaisingEvents = true;
                    p.Exited += (o, e) => 
                    { 
                        Interlocked.Decrement(ref counter); 
                        stopper.Set();

                        StringBuilder stop = new StringBuilder(dic[o as Process]);
                        stop.AppendLine("Stopped:" + DateTime.UtcNow.ToString("s"));
                        dic[o as Process] = stop.ToString();
                    };

                    dic[p] =  cmdline + Environment.NewLine + "Started:" + DateTime.UtcNow.ToString("s");                    
                    Thread.Sleep(pause);
                }
                while (!dic.Keys.All(x => x.HasExited))
                    Thread.Sleep(500);
            }
            finally
            {   
                foreach(var kv in dic)
                {
                    log.AppendLine(kv.Value);
                    kv.Key.Dispose();
                }
            }

            return log.ToString();
        }
    } 
}
